package com.cn.zmc.design.chian.disruptor.config;


import com.cn.zmc.design.chian.disruptor.consumer.ConsumerOne;
import com.cn.zmc.design.chian.disruptor.consumer.ConsumerThree;
import com.cn.zmc.design.chian.disruptor.consumer.ConsumerTwo;
import com.cn.zmc.design.chian.disruptor.handle.*;
import com.cn.zmc.design.chian.domain.Trade;
import com.cn.zmc.design.chian.disruptor.exception.EventExceptionHandler;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.Executor;

@Configuration
public class DisruptorConfig {

    @Autowired
    @Qualifier("systemTaskExecutor")
    private Executor taskExecutor;

    @Autowired
    private HandlerOneService handlerOneService;

    @Autowired
    private HandlerTwoService handlerTwoService;

    @Autowired
    private HandlerThreeService handlerThreeService;

    @Autowired
    private HandlerFourService handlerFourService;

    @Autowired
    private HandlerFiveService handlerFiveService;


   /* * 链式编程，注意taskExecutor：
    *                           todo 线程数 必须大于等于 任务数，否则会卡死住
    * */
    @Bean("tradeRingBuffer")
    public RingBuffer<Trade> tradeRingBuffer() {
        Disruptor<Trade> disruptor = new Disruptor<>(
                new EventFactory<Trade>() {
                    public Trade newInstance() {
                        return new Trade();
                    }
                },
                1024 * 1024,
                taskExecutor,
                ProducerType.SINGLE,
                new BusySpinWaitStrategy());
        //执行顺序（1，4）并发，都执行完成；然后2、5， 2和5都执行完就执行3
        disruptor.handleEventsWith(handlerOneService, handlerFourService);
        disruptor.after(handlerOneService).handleEventsWith(handlerTwoService);
        disruptor.after(handlerFourService).handleEventsWith(handlerFiveService);
        disruptor.after(handlerTwoService, handlerFiveService).handleEventsWith(handlerThreeService);
        disruptor.start();
        return disruptor.getRingBuffer();
    }

    @Autowired
    private ConsumerOne consumerOne;
    @Autowired
    private ConsumerTwo consumerTwo;
    @Autowired
    private ConsumerThree consumerThree;


    @Bean("tradeTwoRingBuffer")
    public RingBuffer<Trade> tradeTwoRingBuffer() {
        //1 创建RingBuffer
        RingBuffer<Trade> ringBuffer =
                RingBuffer.create(ProducerType.MULTI,
                        new EventFactory<Trade>() {
                            public Trade newInstance() {
                                return new Trade();
                            }
                        },
                        1024*1024,
                        new YieldingWaitStrategy());

        //2 通过ringBuffer 创建一个屏障
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        //4 构建多消费者工作池
        WorkerPool<Trade> workerPool = new WorkerPool<Trade>(
                ringBuffer,
                sequenceBarrier,
                new EventExceptionHandler(),
                consumerOne, consumerTwo, consumerThree); //3、创建多个消费者数组

        //5 设置多个消费者的sequence序号 用于单独统计消费进度, 并且设置到ringbuffer中
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
        //6 启动workerPool,配置线程池
        workerPool.start(taskExecutor);
        return ringBuffer;
    }


}
